﻿using System;
using System.Collections.Generic;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.ServiceModel.Description;
using XchangeStreamer.Core.Server;
using System.Threading.Tasks;
using XchangeStreamer.Core;

namespace XchangeStreamer.Server.Core
{
    internal sealed partial class Engine
    {
        #region Private Fields

        private AsyncCallback onNotifyCompleted;
        private int totalSentMessaged = 0;

        #endregion

        #region Public Const.

        public Engine()
        {
            this.sync = Synchronizer.GetSynchronizerInstance(System.Threading.LockRecursionPolicy.NoRecursion);
            this.onNotifyCompleted = new AsyncCallback(NotifyCompleted);
        }

        #endregion

        #region Private Methods

        private void NotifyCompleted(IAsyncResult result)
        {
            //((INotification)(result.AsyncState)).EndNotify(result);
            //((System.ServiceModel.Channels.Message)result.AsyncState).Close();
        }

        private void Notify(System.ServiceModel.Channels.Message message)
        {
            List<INotification> callbacks;
            var topicId = message.Headers.GetHeader<int>(1);
            
            using (new ReadOnlySynchronizer(this.sync, LockType.ReaderWriterLockSlim))
            {
                if (this.subscribers.TryGetValue(topicId, out callbacks))
                {
                    if (callbacks.Count > 0)
                    {
                        MessageBuffer messageBuffer = message.CreateBufferedCopy(65536);
                        //message.Close();

                        for (int i = 0; i < callbacks.Count; i++)
                        {
                            var msg = messageBuffer.CreateMessage();
                            callbacks[i].BeginNotify(msg, null, null);
                            System.Threading.Interlocked.Increment(ref this.totalSentMessaged);
                        }
                        //Parallel.For(0, callbacks.Count, i =>
                        //{
                        //    var msg = messageBuffer.CreateMessage();
                        //    callbacks[i].BeginNotify(msg, null, null);
                        //    System.Threading.Interlocked.Increment(ref this.totalSentMessaged);
                        //});

                        messageBuffer.Close();
                    }
                    else
                    {
                        //message.Close();
                    }
                }
                else
                {
                    //message.Close();
                }   
            }
        }

        #endregion

        #region Public Prop.

        public int TotalSentMesaages
        {
            get
            {
                return this.totalSentMessaged;
            }
        }

        #endregion
    }
}
